package fi.aalto.cse.T110_5150.P2PClient;

import java.io.*;
import java.net.*;
import java.util.*;

public class RemotePeer implements Runnable {

	private PeerAddress remotePeerAddress;
	private Socket socket;
	private DataOutputStream out;
	private DataInputStream in;
	private boolean isInitialized;
	private boolean sendJoinRequest;
	private boolean isStopped;
	private long lastHeartBeatTime;
	private long lastNetworkProbingTime;
	private long waitingForInputMessageTime;

	public RemotePeer(Socket socket, PeerAddress remotePeerAddress, boolean sendJoinRequest) throws IOException {
		this.socket = socket;
		this.remotePeerAddress = remotePeerAddress;
		this.sendJoinRequest = sendJoinRequest;

		isInitialized = false;
		out = new DataOutputStream(socket.getOutputStream());
		in = new DataInputStream(socket.getInputStream());
	}

	@Override
	public void run() {

		try {
			if (sendJoinRequest) {
				sendMessageJoin();
			} else {
				setWaitingForResponse(true);
			}
			
			int count = 0;

			while (!isStopped && !socket.isClosed()) {
				if (in.available() > 0) {
					count = 0;
					try {
						handleMessage();
					} catch (IOException e) {
						// ignore the exception only if the peer is requested to
						// stop
						if (!isStopped) {
							throw e;
						}
					}
				} else {
					if (count < 100) {
						Thread.sleep(10);
						count++;
					} else {
						Thread.sleep(100);
					}

					if (!isStopped && in.available() == 0) {

						if (waitingForInputMessageTime != 0) {

							if (isExpired(waitingForInputMessageTime, P2PHelper.SOCKET_INPUT_TIMEOUT)) {
								throw new P2PException("timeout is exceeded");
							}

						} else if (isInitialized) {

							if (LocalPeer.getInstance().getAutoJoin() &&
									isExpired(lastNetworkProbingTime, P2PHelper.PING_PONG_NETWORK_PROBING_INTERVAL)) {
								lastNetworkProbingTime = System.currentTimeMillis();
								sendMessagePing(P2PHelper.PING_PONG_NETWORK_PROBING_TTL);
							} else if (isExpired(lastHeartBeatTime, P2PHelper.PING_PONG_HEART_BEAT_INTERVAL)) {
								lastHeartBeatTime = System.currentTimeMillis();
								sendMessagePing(P2PHelper.PING_PONG_HEART_BEAT_TTL);
							}

						}

					}
				}
			}

			if (isStopped && !socket.isClosed()) {
				sendMessageBye();
				Logger.infoOutputAction("DISCONNECTED (by this peer)", remotePeerAddress);
			} else {
				Logger.infoOutputAction("DISCONNECTED (by the remote peer)", remotePeerAddress);
			}

		} catch (P2PException e) {
			Logger.warningInputAction(String.format("DISCONNECTED (%s)", e.getMessage()), remotePeerAddress);			
		} catch (IOException e) {
			Logger.warningInputAction(String.format("DISCONNECTED (I/O error: %s)", e.getMessage()), remotePeerAddress);
			//Logger.error(e);			
		} catch (Exception e) {
			System.out.printf("Unexpected error: %s.%n", e.getMessage());
			e.printStackTrace();
		} finally {
			finalize();
			try {
				LocalPeer.getInstance().removeRemotePeer(this);
			} catch (P2PException e) {
//				Logger.error(e);
			}
		}
	}

	public boolean getInitializedStatus() {
		return isInitialized;
	}

	public PeerAddress getAddress() {
		return remotePeerAddress;
	}

	public void stop(boolean waitForStop) {
		isStopped = true;
	}

	@Override
	protected void finalize() {

		try {
			super.finalize();

			isStopped = true;
			isInitialized = false;

			if (socket != null) {
				socket.close();
			}

			// stoppedEvent.notifyAll();
		} catch (Throwable e) {
			Logger.error("Unexpected error", e);
		} finally {
			socket = null;
		}
	}

	private void sendHeader(MessageHeader header) throws IOException {
		out.writeByte(header.version);
		out.writeByte(header.ttl);
		out.writeByte(header.messageType);
		out.writeByte(header.reserved);

		out.writeShort(header.originalPort);
		out.writeShort(header.length);

		out.write(header.originalAddress);

		out.writeInt(header.messageId);
	}

	private void handleMessage() throws IOException, P2PException {
		// read header
		MessageHeader header = new MessageHeader();
		header.version = in.readUnsignedByte();
		header.ttl = in.readUnsignedByte();
		header.messageType = in.readUnsignedByte();
		header.reserved = in.readUnsignedByte();

		header.originalPort = in.readUnsignedShort();
		header.length = in.readUnsignedShort();

		in.readFully(header.originalAddress);
		header.messageId = in.readInt();

		if (header.version == P2PHelper.VERSION_CURRENT) {
			switch (header.messageType) {
			case P2PHelper.MESSAGE_TYPE_JOIN:
				handleMessageJoin(header);
				break;
			case P2PHelper.MESSAGE_TYPE_PING:
				handleMessagePing(header);
				break;
			case P2PHelper.MESSAGE_TYPE_PONG:
				handleMessagePong(header);
				break;
			case P2PHelper.MESSAGE_TYPE_BYE:
				handleMessageBye(header);
				break;
			case P2PHelper.MESSAGE_TYPE_QUERY:
				handleMessageQuery(header);
				break;
			case P2PHelper.MESSAGE_TYPE_QUERY_HIT:
				handleMessageQueryHit(header);
				break;

			default:
				// unsupported message type
				String message = String.format("Unsupported message type %d (message ID = %d)", header.messageType,
						header.messageId);
				Logger.warningInputAction(message, remotePeerAddress);
				break;
			}
		} else {
			// incompatible version
			in.skipBytes(header.length);
			Logger.warningInputAction(String.format("Incompatible version (skipping message) received version = %d", header.version), remotePeerAddress);
		}
	}

	private void sendMessageJoin() throws IOException {
		MessageHeader header = MessageHeader.create(P2PHelper.MESSAGE_TYPE_JOIN, 0, 0);
		sendHeader(header);

		Logger.debugOutputAction(Logger.JOIN, "Sent Join Request", remotePeerAddress);
		setWaitingForResponse(true);
	}

	private void handleMessageJoin(MessageHeader header) throws IOException, P2PException {
		switch (header.length) {
		case 0:
			// received JoinRequest
			remotePeerAddress = new PeerAddress(InetAddress.getByAddress(header.originalAddress), header.originalPort);
			isInitialized = true;
			Logger.debugInputAction(Logger.JOIN, "Received Join Request", remotePeerAddress);

			// send JoinResponse
			header.originalPort = LocalPeer.getInstance().getAddress().getPort();
			header.length = 2;
			header.originalAddress = LocalPeer.getInstance().getAddress().getAddressBytes();
			sendHeader(header);

			out.writeShort(P2PHelper.STATUS_OK);

			Logger.debugOutputAction(Logger.JOIN, "Sent Join Response: OK", remotePeerAddress);
			Logger.infoInputAction("CONNECTED (initialized by the remote node)", remotePeerAddress);

			break;
		case 2:
			// received Join Response

			int status = in.readUnsignedShort();

			if (status == P2PHelper.STATUS_OK) {
				isInitialized = true;
				Logger.debugInputAction(Logger.JOIN, "Received Join Response: OK", remotePeerAddress);
				Logger.infoOutputAction("CONNECTED (initialized by this peer)", remotePeerAddress);
			} else {
				Logger.warningInputAction("Received Join Response: Rejected!", remotePeerAddress);
				isStopped = true;
			}

			break;
		}

		if (isInitialized) {
			if (!sendJoinRequest) {
				LocalPeer.getInstance().addRemotePeer(this);
			}
			lastHeartBeatTime = System.currentTimeMillis();
		}

		setWaitingForResponse(false);
	}

	private void sendMessagePing(int ttl) throws IOException {
		MessageHeader header = MessageHeader.create(P2PHelper.MESSAGE_TYPE_PING, ttl, 0);
		sendHeader(header);

		if (ttl != P2PHelper.PING_PONG_HEART_BEAT_TTL) {
			Logger.debugOutputAction(Logger.NETWORK_PROBING, String.format("Sent Ping (TTL=%d)", ttl), remotePeerAddress);
		}
	}

	private void handleMessagePing(MessageHeader header) throws IOException {
		if (header.ttl != P2PHelper.PING_PONG_HEART_BEAT_TTL) {
			Logger.debugInputAction(Logger.HEART_BEAT, String.format("Received Ping (TTL=%d)", header.ttl), remotePeerAddress);
		}
		header.messageType = P2PHelper.MESSAGE_TYPE_PONG;

		header.originalPort = LocalPeer.getInstance().getAddress().getPort();
		header.originalAddress = LocalPeer.getInstance().getAddress().getAddressBytes();

		if (header.ttl == P2PHelper.PING_PONG_HEART_BEAT_TTL) {
			sendHeader(header);
			Logger.debugOutputAction(Logger.HEART_BEAT, String.format("Sent Pong (TTL=%d)", header.ttl), remotePeerAddress);
		} else {
			List<PeerAddress> randomAddresses = LocalPeer.getInstance().getRandomRemotePeerAddresses(remotePeerAddress);

			int numberOfAddresses = randomAddresses.size();
			header.length = numberOfAddresses * 8 + 4;

			sendHeader(header);
			out.writeShort(numberOfAddresses);
			out.writeShort(0);

			for (Iterator<PeerAddress> it = randomAddresses.iterator(); it.hasNext(); ) {
				PeerAddress randomAddress = it.next();
				out.write(randomAddress.getAddressBytes());
				out.writeShort(randomAddress.getPort());
				out.writeShort(0);
			}

			Logger.debugOutputAction(Logger.NETWORK_PROBING, String.format("Sent Pong (TTL=%d, peers=%d)", header.ttl, numberOfAddresses),
					remotePeerAddress);
		}
	}

	private void handleMessagePong(MessageHeader header) throws IOException {
		if (header.ttl == P2PHelper.PING_PONG_HEART_BEAT_TTL) {
			Logger.debugInputAction(Logger.HEART_BEAT, String.format("Received Pong (TTL=%d)", header.ttl), remotePeerAddress);
		} else {
			int numberOfPeers = in.readUnsignedShort();
			in.skipBytes(2);

			for (int i = 0; i < numberOfPeers; ++i) {
				byte[] address = new byte[4];
				in.readFully(address, 0, 4);
				int port = in.readUnsignedShort();
				in.skipBytes(2);
				
				if (LocalPeer.getInstance().getAutoJoin()) {
					PeerAddress newRemotePeerAddress = new PeerAddress(InetAddress.getByAddress(address), port);
					try {
						LocalPeer.getInstance().connect(newRemotePeerAddress);
					} catch (P2PAlreadyConnectedException e) {						
					} catch (P2PException e) {
						Logger.error(e.getMessage());
					}
				}
			}

			Logger.debugInputAction(Logger.NETWORK_PROBING, String.format("Received Pong (TTL=%d, peers=%d)", header.ttl, numberOfPeers),
					remotePeerAddress);
			lastNetworkProbingTime = System.currentTimeMillis();
		}
	}

	private void sendMessageBye() throws IOException {
		MessageHeader header = MessageHeader.create(P2PHelper.MESSAGE_TYPE_BYE, 1, 0);
		sendHeader(header);
		Logger.debugOutputAction(Logger.JOIN, "Sent Bye Message", remotePeerAddress);

		while (!socket.isClosed() && socket.isConnected()) {
			try {
				socket.setSoTimeout(1);
				in.read();
			} catch (IOException e) {
				break;
			}
		}
	}

	private void handleMessageBye(MessageHeader header) throws IOException {
		Logger.debugOutputAction(Logger.JOIN, "Received Bye Message", remotePeerAddress);
		socket.close();
	}

	private void handleMessageQuery(MessageHeader header) throws IOException {
		
		/* read query array */
		byte[] buf = new byte[header.length]; 
		in.readFully(buf);		
		ByteArray queryKey = new ByteArray(buf);
		LocalPeer.getInstance().handleMessageQuery(header, queryKey, this);
	}

	public void handleMessageQueryHit(MessageHeader header) throws IOException{
		
		/* read entry size, resource id and resource value */
		int numberOfEntries = in.readShort();
		in.skipBytes(2);
		
		Resource[] resources = new Resource[numberOfEntries];
		
		for (int i = 0; i < numberOfEntries; ++i) {
			short resourceId = in.readShort();
			in.skipBytes(2);
			int resourceValue = in.readInt();
			
			resources[i] = new Resource(resourceId, resourceValue);
		}
		
		LocalPeer.getInstance().handleMessageQueryHit(header, resources);		
	}	

	public void sendMessageQuery(MessageHeader header, ByteArray queryKey) throws IOException {		
		sendHeader(header);

		// send query array
		out.write(queryKey.toByteArray());			

		Logger.debugOutputAction(Logger.QUERY, "Sent Query Message", remotePeerAddress);
	}

	public void sendMessageQueryHit(int messageId, Resource[] resources) throws IOException {
		int size = 4 + resources.length * 8; 
		
		MessageHeader header = MessageHeader.create(messageId, P2PHelper.MESSAGE_TYPE_QUERY_HIT,
				P2PHelper.QUERY_MAX_TTL, size);
		
		sendHeader(header);
		out.writeShort(resources.length);
		out.writeShort(0);
		
		for (int i = 0; i < resources.length; ++i) {
			out.writeShort(resources[i].getId());
			out.writeShort(0);
			out.writeInt(resources[i].getValue());
		}
	}

	private void setWaitingForResponse(boolean waitForResponse) throws SocketException {
		if (waitForResponse) {
			Logger.debugOutputAction(Logger.JOIN, "Waiting for response", remotePeerAddress);
			waitingForInputMessageTime = System.currentTimeMillis();
		} else {
			waitingForInputMessageTime = 0;
		}
	}

	private static boolean isExpired(long time, long expiringInterval) {
		long currentTime = System.currentTimeMillis();
		return time + expiringInterval < currentTime;
	}

}
